/**
 * 
 */
package com.rrd.dw.mr.gzip;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.MD5Util;

/**
 * @author xiexihao
 */
public class RongRecordsExtractor extends Configured implements Tool {

    private static final String STR_FILED_SPLIT_FLAG = "\001";
    private static final String STR_TB_PREFIX_FLAG = "\007";

    private static final String STR_TARGET_POSITION_ORDER_BASIC_INF = "orderbasicinfo";
    private static final String STR_TARGET_POSITION_ORDER_ADDITIONAL_INF = "orderadditionalinfo";
    private static final String STR_TARGET_POSITION_ORDER_JOB_INF = "orderjobinfo";

    private static final byte[] FAMILY_INF = Bytes.toBytes("inf");
    private static final byte[] FAMILY_JOB = Bytes.toBytes("job");
    private static final byte[] COLUMN_INF_BASIC_INFO = Bytes
        .toBytes("base_order_info");
    private static final byte[] COLUMN_INF_ADDITIONAL_INFO = Bytes
        .toBytes("additional_order_info");

    private static class RongRecordsMapper extends TableMapper<Text, Text> {
        private Text key = null;
        private Text value = null;

        protected void setup(Context context) {
            key = new Text();
            value = new Text();
        }

        protected void map(ImmutableBytesWritable key, Result value,
                Context context) throws IOException, InterruptedException {
            String keyStr = null;
            StringBuffer keysb = new StringBuffer();

            keysb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("job_id"))));
            keysb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("system_id")) == null
                    ? ""
                    : Bytes.toString(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("system_id"))));
            keysb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("type")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("type"))));
            keyStr = MD5Util.getMd5(keysb.toString());
            this.key.set(keyStr);

            StringBuffer sb = new StringBuffer();
            sb.append(STR_TARGET_POSITION_ORDER_JOB_INF + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("job_id")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("job_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("system_id")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("system_id"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("type")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("type"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("try_times")) == null
                    ? ""
                    : String.valueOf(Bytes.toInt(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("try_times")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("timestamp")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("timestamp")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("successful")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("successful")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("start_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("start_time")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("requested")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("requested")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("query")) == null ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("query"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("message")) == null
                    ? ""
                    : Bytes.toString(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("message"))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("finished")) == null
                    ? ""
                    : String.valueOf(Bytes.toBoolean(
                        value.getValue(FAMILY_JOB, Bytes.toBytes("finished")))))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                value.getValue(FAMILY_JOB, Bytes.toBytes("end_time")) == null
                    ? ""
                    : String.valueOf(Bytes.toLong(value.getValue(FAMILY_JOB,
                        Bytes.toBytes("end_time")))));

            this.value.set(sb.toString());

            context.write(this.key, this.value);
            parseOrderBasicInfo(context, keyStr,
                value.getValue(FAMILY_INF, COLUMN_INF_BASIC_INFO));
            parseOrderAdditionalInfo(context, keyStr,
                value.getValue(FAMILY_INF, COLUMN_INF_ADDITIONAL_INFO));
        }

        private void parseOrderBasicInfo(Context context, String keyStr,
                byte[] orderBasicInfo)
                throws IOException, InterruptedException {
            if (orderBasicInfo == null)
                return;

            StringBuffer sb = new StringBuffer();

            JSONObject basicInfo = JSON
                .parseObject(Bytes.toString(orderBasicInfo));
            JSONObject orderInfo = basicInfo.getJSONObject("orderInfo");
            JSONObject applyDetail = basicInfo.getJSONObject("applyDetail");
            JSONObject addInfo = basicInfo.getJSONObject("addInfo");
            JSONObject zhimaScore = addInfo != null
                && addInfo.getJSONObject("zhima") != null
                    ? addInfo.getJSONObject("zhima").getJSONObject(
                        "ali_trust_score")
                    : null;
            JSONObject contacts = addInfo == null ? null
                : addInfo.getJSONObject("contacts");
            JSONObject gpsInfo = contacts == null ? null
                : contacts.getJSONObject("app_location");

            sb.append(STR_TARGET_POSITION_ORDER_BASIC_INF + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("order_no"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("user_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("user_mobile"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("user_id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("application_amount"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("application_term"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("order_time"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("status"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("city")).append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("bank")).append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("product"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("product_id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(orderInfo.getString("platform"))
                .append(STR_FILED_SPLIT_FLAG);

            sb.append(applyDetail.getString("is_op_type"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("user_id_card_no"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("loan_apply_term"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("asset_auto_type"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("max_monthly_repayment"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("user_income_by_card"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("user_education"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("user_social_security"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("bureau_user_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("work_period"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("phone_number_house"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("platform"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("device_num"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("ip_address"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("tele_num"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("tele_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(applyDetail.getString("phone_brand"))
                .append(STR_FILED_SPLIT_FLAG);

            String[] gpsStr = applyDetail.getString("gps_location") == null
                ? null
                : applyDetail.getString("gps_location").split(",");
            sb.append(gpsStr != null && gpsStr.length > 0 && gpsStr[0] != null
                ? gpsStr[0]
                : gpsInfo == null ? "" : gpsInfo.getString("lat"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(gpsStr != null && gpsStr.length > 1 && gpsStr[1] != null
                ? gpsStr[1]
                : gpsInfo == null ? "" : gpsInfo.getString("lon"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(gpsInfo != null && gpsInfo.getString("address") != null
                ? gpsInfo.getString("address")
                : applyDetail.getString("gps_Address"))
                .append(STR_FILED_SPLIT_FLAG);

            sb.append(zhimaScore == null ? "" : zhimaScore.getString("score"))
                .append(STR_FILED_SPLIT_FLAG);

            sb.append(contacts == null ? "" : contacts.getString("id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("uid"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("device_info"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                contacts == null || contacts.getString("installed_apps") == null
                    ? ""
                    : contacts.getString("installed_apps")
                        .replaceAll("\r\n|\t|\n|\r", ""))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? ""
                : contacts.getString("installed_apps_version"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("update_time"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("create_time"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                contacts == null ? "" : contacts.getString("delete_time"));

            this.key.set(keyStr);
            this.value.set(sb.toString());

            context.write(this.key, this.value);
        }

        private void parseOrderAdditionalInfo(Context context, String keyStr,
                byte[] orderAdditionalInfo)
                throws IOException, InterruptedException {
            if (orderAdditionalInfo == null)
                return;

            StringBuffer sb = new StringBuffer();

            JSONObject additionalInfo = JSON
                .parseObject(Bytes.toString(orderAdditionalInfo));
            JSONObject contacts = additionalInfo.getJSONObject("contacts");
            sb.append(
                STR_TARGET_POSITION_ORDER_ADDITIONAL_INF + STR_TB_PREFIX_FLAG);
            sb.append(keyStr).append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("order_no"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("user_id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("product_id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("app_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("app_version"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("channel"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("platform"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("Name_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ID_Sex_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ID_Ethnic_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ID_Address_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ID_Number_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ID_Issue_Org_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ID_Due_time_OCR"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("scores_assay"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("company_addr_detail"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("company_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo
                .getString("emergency_contact_personA_relationship"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                additionalInfo.getString("emergency_contact_personA_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                additionalInfo.getString("emergency_contact_personA_phone"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo
                .getString("emergency_contact_personB_relationship"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                additionalInfo.getString("emergency_contact_personB_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                additionalInfo.getString("emergency_contact_personB_phone "))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("familiy_live_type"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("work_industry"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("photo_assay_time"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("family_live_year"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("user_marriage_simple"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("living_city"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("work_position"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("is_simulator"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("amount_of_staff"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("time_enrollment"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("school_name"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("ip_address"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("user_email"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(additionalInfo.getString("addr_detail"))
                .append(STR_FILED_SPLIT_FLAG);

            sb.append(contacts == null ? "" : contacts.getString("id"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("uid"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("update_time"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(contacts == null ? "" : contacts.getString("create_time"))
                .append(STR_FILED_SPLIT_FLAG);
            sb.append(
                contacts == null ? "" : contacts.getString("delete_time"));

            this.key.set(keyStr);
            this.value.set(sb.toString());

            context.write(this.key, this.value);
        }
    }

    private static class RongRecordsReducer
            extends Reducer<Text, Text, Text, Text> {
        private NullWritable key2 = NullWritable.get();
        private MultipleOutputs<Text, Text> mos = null;

        protected void setup(Context context)
                throws IOException, InterruptedException {
            mos = new MultipleOutputs<Text, Text>(context);
        }

        protected void cleanup(Context context)
                throws IOException, InterruptedException {
            mos.close();
        }

        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text value : values) {
                String[] contents = value.toString().split(STR_TB_PREFIX_FLAG);
                if (contents == null)
                    continue;

                if (STR_TARGET_POSITION_ORDER_BASIC_INF.equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_ORDER_BASIC_INF, key2,
                        new Text(contents[1]));
                } else if (STR_TARGET_POSITION_ORDER_ADDITIONAL_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_ORDER_ADDITIONAL_INF,
                        key2, new Text(contents[1]));
                } else if (STR_TARGET_POSITION_ORDER_JOB_INF
                    .equals(contents[0])) {
                    this.mos.write(STR_TARGET_POSITION_ORDER_JOB_INF, key2,
                        new Text(contents[1]));
                }
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        // TODO Auto-generated method stub
        Properties prop = new Properties();
        try (InputStream is = this.getClass().getClassLoader()
            .getResourceAsStream("hbase.properties");) {
            prop.load(is);
        }
        Job job = Job.getInstance(HBaseConfiguration.create(this.getConf()));
        Configuration conf = job.getConfiguration();
        conf.set("hbase.zookeeper.quorum", "datanode1,datanode2,datanode3");
        conf.set("mapreduce.map.memory.mb", "3072");
        conf.set("mapreduce.reduce.memory.mb", "5120");
        conf.set("hbase.client.scanner.timeout.period", "120");
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec",
            "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs",
            "org.apache.hadoop.io.compress.GzipCodec");

        /*
         * conf.set("param.out.path",
         * "hdfs://10.10.15.10:8020/tmp/output_xxh_rongreport/");
         * conf.set("param.start_date", "2017-09-19");
         * conf.set("param.stop_date", "2017-09-20");
         */
        conf.set("param.out.path", args[0]);
        conf.set("param.start_date", args[1]);
        conf.set("param.stop_date", args[2]);

        String tableName = "rong_records";
        String outputDir = conf.get("param.out.path");
        System.out.println("开始日期:" + conf.get("param.start_date"));
        System.out.println("结束日期:" + conf.get("param.stop_date"));
        System.out.println("输入HBase表名:" + tableName);
        System.out.println("输出路径:" + outputDir);

        Long startTime = DateUtil
            .formateToTimestamp(conf.get("param.start_date"));
        Long endTime = DateUtil.formateToTimestamp(conf.get("param.stop_date"));

        Path outpath = new Path(outputDir);
        FileSystem fs = outpath.getFileSystem(conf);
        if (fs.exists(outpath)) {
            fs.delete(outpath, true);
        }

        job.setJarByClass(RongReportsExtractor.class);
        Scan scan = new Scan();

        scan.addFamily(FAMILY_INF);
        scan.addColumn(FAMILY_INF, COLUMN_INF_BASIC_INFO);
        scan.addColumn(FAMILY_INF, COLUMN_INF_ADDITIONAL_INFO);
        scan.addFamily(FAMILY_JOB);

//		scan.setCacheBlocks(false);
//		scan.setCaching(200);
//		scan.setBatch(100);
        scan.setMaxVersions(1);
//		scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes(tableName));
        scan.setTimeRange(startTime, endTime + 14400000);

        TableMapReduceUtil.initTableMapperJob(tableName, scan,
            RongRecordsMapper.class, Text.class, Text.class, job);
        FileOutputFormat.setOutputPath(job, outpath);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_ORDER_BASIC_INF,
            TextOutputFormat.class, Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job,
            STR_TARGET_POSITION_ORDER_ADDITIONAL_INF, TextOutputFormat.class,
            Text.class, Text.class);
        MultipleOutputs.addNamedOutput(job, STR_TARGET_POSITION_ORDER_JOB_INF,
            TextOutputFormat.class, Text.class, Text.class);

        job.setReducerClass(RongRecordsReducer.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    /**
     * @param args
     */
    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        System.exit(ToolRunner.run(HBaseConfiguration.create(),
            new RongRecordsExtractor(), args));
    }
}
